package com.yl.rabbitmq.day05;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.yl.rabbitmq.utils.RabbitMQUtil;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @author yangl
 * 发布确认模式
 * 1、单个确认
 * 2、批量确认
 * 3、异步确认
 * 比较使用的时间，比较哪种确认方式最好
 */
public class Producer05 {

    /**
     * 批量发送消息的个数
     * @param args
     */
    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {
        //1、单个确认
//        Producer05.publishMessageSingle();
        //2、批量确认
//        Producer05.publishMessageBatch();
        //3、异步批量确认
        Producer05.publishMessageAsync();
    }

    public static void publishMessageSingle() throws Exception{
        Channel channel = RabbitMQUtil.getChannel();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        //开始时间
        //开启发布确认
        channel.confirmSelect();
        long start = System.currentTimeMillis();
        //批量发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());
            //单个消息发送之后，马上进行发布确认
            boolean b = channel.waitForConfirms();
            if(b){
                System.out.println("消息发送成功");
            }
        }
        long end = System.currentTimeMillis();
        long l = end - start;
        System.out.println("发送" + MESSAGE_COUNT + "个消息，单独确认消耗的时间是：" + l);
    }

    public static void publishMessageBatch() throws Exception{
        Channel channel = RabbitMQUtil.getChannel();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        //开始时间
        //开启发布确认
        channel.confirmSelect();
        long start = System.currentTimeMillis();
        //批量确认消息的大小
        int batchSize = 100;
        //批量发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());
            //判断达到100条消息的时候，确认一次
            if(i % 100 == 0){
                channel.waitForConfirms();
            }
        }
        long end = System.currentTimeMillis();
        long l = end - start;
        System.out.println("发送" + MESSAGE_COUNT + "个消息，批量确认消耗的时间是：" + l);
    }

    public static void publishMessageAsync() throws Exception{
        Channel channel = RabbitMQUtil.getChannel();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);

        //开启发布确认
        channel.confirmSelect();
        /**
         * 线程安全有序的hash表，适用于高并发的情况下
         * 1、轻松的将序号与消息进行关联
         * 2、轻松的批量删除条目，只要给到序号
         * 3、支持高并发（多线程）
         */
        ConcurrentSkipListMap<Long, String> concurrentSkipListMap = new ConcurrentSkipListMap<>();

        //消息确认成功，回调
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            //2、删除已确认的消息，剩下的就是未确认的消息
            if(multiple){
                ConcurrentNavigableMap<Long, String> confirmed = concurrentSkipListMap.headMap(deliveryTag);
                confirmed.clear();
            }else{
                concurrentSkipListMap.remove(deliveryTag);
            }

            System.out.println("确认的消息" + deliveryTag);
        };
        /**
         * 消息确认失败，回调
         * 1、消息的标记
         * 2、是否为批量确认
         */

        ConfirmCallback nackCallBack = (deliveryTag, multiple) -> {
            //3、打印未确认的消息
            String message = concurrentSkipListMap.get(deliveryTag);
            System.out.println("未确认的消息:" + message + ", 未确认的消息的标记是:" + deliveryTag);
        };
        /**
         * 消息监听器，来监听消息是否成功，从而知道哪些消息需要重发
         * 1、监听哪些消息成功了
         * 2、监听哪些消息失败了
         * 一个参数是监听成功的
         * 两个参数：一个是监听成功，一个是监听失败的
         */
        channel.addConfirmListener(ackCallback, nackCallBack);
        //开始时间
        long start = System.currentTimeMillis();
        //批量发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());
            //1、此处记录所有要发送的消息
            concurrentSkipListMap.put(channel.getNextPublishSeqNo(),message);
        }
        
        long end = System.currentTimeMillis();
        long l = end - start;
        System.out.println("发送" + MESSAGE_COUNT + "个消息，异步确认消耗的时间是：" + l);
    }
}
